package com.zhou;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @Description: kafka练习
 * @Author: ZhOu
 * @Date: 2018/7/19
 */
public class KafkaTest {

    /**
     * 创建主题
     */
    @Test
    public void createTopic() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "192.168.66.14:9092");
        AdminClient client = AdminClient.create(properties);
        ArrayList<NewTopic> topics = new ArrayList<NewTopic>();
        NewTopic topic1 = new NewTopic("topic-test1", 1, (short) 1);
        NewTopic topic2 = new NewTopic("topic-test2", 1, (short) 1);
        topics.add(topic1);
        topics.add(topic2);
        CreateTopicsResult result = client.createTopics(topics);

        try {
            result.all().get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建生产者
     */
    @Test
    public void createProducer() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "192.168.66.14:9092");
        properties.put("acks", "all");
        properties.put("reties", 0);
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 102400);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<String, String>("test", Integer.toString(i), "这是第" + i + "个消息"));
        }
        producer.close();
    }

    /**
     * 消费消息
     */
    @Test
    public void createConsumer() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "192.168.66.14:9092");
        properties.put("group.id", "test");
        properties.put("enable.auto.commit", "true");
        properties.put("auto.commit.interval.ms", "1000");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        consumer.subscribe(Arrays.asList("topic-test", "test"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> collection) {

            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                //设置偏移量
                consumer.seekToBeginning(collection);
            }
        });

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            records.forEach(m -> {
                System.out.printf("offset=%d, key=%s, value=%s%n", m.offset(), m.key(), m.value());
                System.out.println(m.toString());
            });
        }

    }
}
